package com.shujia.core.transformations.kv

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object GroupByKeyOpt {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("groupByKey算子操作")

    val sc = new SparkContext(conf)

    /**
     *  今后我们在spark中看到xxxXxxByKey,这样的算子大多数只能作用于KV类的RDD上  RDD[(key,value)]
     *
     *  自动将相同的键为一组，不需要手动指定分组的键
     */
    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    //需求：计算每个班级的平均年龄
    val clazzWithAgeRDD: RDD[(String, Int)] = lineRDD.map((line: String) => {
      val infos: Array[String] = line.split(",")
      (infos(4),infos(2).toInt)
    })

    /**
     * 面试题：groupBy和groupByKey的区别
     *
     * 1、groupBy自定义指定分组的条件，任意一个RDD都可以调用；而groupByKey只能由KV类型的RDD才能调用，自动根据键进行分组
     * 2、groupBy分组后的结果，是将元素本身整体放入到一个迭代器中；而groupByKey分组后的结果，是只将值放入到一个迭代器中
     * 3、groupByKey中键产生的shuffle数据量要远小于groupBy中国产生的shuffle数据量。所以当前一个RDD是KV类型且需要分组的时候，推荐使用groupByKey。效率要更高一些，速度快一些。
     * 4、groupBy的使用灵活度要高于groupByKey
     *
     */
    val clazzGroupRDD: RDD[(String, Iterable[Int])] = clazzWithAgeRDD.groupByKey()

    val resRDD: RDD[(String, Int)] = clazzGroupRDD.map((kv: (String, Iterable[Int])) => {
      val clazz: String = kv._1
      val avgAge: Int = kv._2.toList.sum / kv._2.toList.size
      (clazz, avgAge)
    })

    resRDD.foreach(println)

    while (true){

    }

  }
}
